/**  
  -- The Performance Driven Lime Project -- 

  A self adaptive tuple space for mobile agents and hosts by dynamic
  adapting tuple's replication policy.
  
  Copyright (C) 2008
  Xing Jiankuan, Pan Wenhui
  SAGroup of Tsinghua University

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
 */
package sagroup.dracots;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;


import lights.adapters.TupleSpace;
import lights.interfaces.ITuple;
import lights.interfaces.TupleSpaceException;

import lime2.TupleID;

import sagroup.dracots.SystemTupleType;
import sagroup.dracots.TupleType;
import sagroup.util.Util;

/**
 * <p>
 * Manage all local replicas. This class contains a LRU based cache and a
 * Replica Usage Info Set. The LRU cache is meant to limit the total capacity of
 * tuples. If a new replica is added and the capacity threshold is touched, the
 * eldest entries will be removed until capacity is less than threshold again.
 * To keep the LRU cache is always the access order, each access to a replica
 * tuple should invoke {@link LocalReplicaRegistrar#touch(TupleID vid)}.
 * </p>
 * <p>
 * The Usage Info Set records each replica's usage, that is, Used or Unused.
 * When the replica is first replicated and add to LocalReplicaRegistrar, it is
 * marked as Unused. When it is first touched, it is remarked as Used. The usage
 * info of tuples of a type is the basic data with which adapter decides to use
 * FR or RC.
 * </p>
 * <p>
 * The replicas with type REPLICATION_POLICY will never be added to this
 * registrar since their policy is always FR.
 * </p>
 * <p>
 * This class is thread-safe.
 * </p>
 * 
 * @author Xing Jiankuan
 * 
 */
public class LocalReplicaRegistrar {

	private DracoTS dts;

	private HashMap map;

	/*
	 * The Hash map of TupleType <-> added count from the beginning
	 */
	private HashMap<TupleType, Long> typeAdded;

	/*
	 * The Hash map of TupleType <-> deleted count from the beginning
	 */
	private HashMap<TupleType, Long> typeDeleted;

	/**
	 * The header of link list, point to a dummy entry.
	 * <code>header.after</code> and <code>header.before</code> point to header
	 * itself means an empty linked list.
	 */
	private transient Entry header;

	/**
	 * Point to the first unused entry, or null if there is no entry in the
	 * list.
	 */
	private transient Entry unusedHeader;

	private long maxCapacity;

	private int currCapacity = 0;

	/**
	 * Replica Tuple Space
	 */
	private TupleSpace replicaTS;

	/**
	 * Create a new LocalReplicaRegistrar by given a capacity upper bound.
	 * 
	 * @param maxCapacity
	 *            the upper bound of replica's capacity. Note this capacity is
	 *            not the count of hash map items, but the total size of tuple
	 *            in byte.
	 */
	public LocalReplicaRegistrar(long maxCapacity, DracoTS dts) {
		map = new HashMap();
		typeAdded = new HashMap();
		typeDeleted = new HashMap();
		this.maxCapacity = maxCapacity;
		this.dts = dts;
		try {
			replicaTS = new TupleSpace(dts.getName() + "_replicas");
		} catch (TupleSpaceException e) {
			
			e.printStackTrace();
		}
		init();
	}

	void init() {
		header = new Entry(null, null, -1);
		header.before = header;
		header.after = header;
	}

	synchronized public boolean containsVID(TupleID vid) {
		return map.containsKey(vid);
	}

	synchronized public Entry get(TupleID vid) {
		return (Entry) map.get(vid);
	}

	public boolean isUsed(TupleID vid) {
		return get(vid).used;
	}

	public TupleType getType(TupleID vid) {
		return get(vid).type;
	}

	/**
	 * Add a replica to local host replica the new replica to be added or not
	 * 
	 * @author Xing Jiankuan
	 */
	synchronized public boolean addReplica(ShareTuple replica) {
		assert (replica.isReplica);

		// check whether the repilca is already cached.
		TupleID vid = replica.vid;
		if (map.containsKey(vid))
			return false;

		if (!SystemTupleType.isSystemTupleType(replica.getType())) {
			// only replicas of non-system tuple type need replica management
			int size = replica.getSize();

			if (size > maxCapacity) {
				System.err.println("Cannot create replica whose size is larger"
						+ " than maximal capacity");
				return false;
			}

			// remove replicas according to LRU
			while (currCapacity + size > maxCapacity) {
				Entry eldest = header.after;
				map.remove(eldest.vid);

				if (eldest == unusedHeader) {
					if (eldest.before == header) // the only one unused entry is
						// removed
						unusedHeader = null;
					else
						unusedHeader = eldest.before;
				}
				eldest.remove();
				decreaseUsedCountByType(eldest);
				rinp(ShareTuple.createVIDTemplate(eldest.vid));
						
				currCapacity -= eldest.size;
			}

			// dts.getPolicyDecisionMaker().handleNewcoming(replica);

			// add it to hash map
			Entry entry = new Entry(replica.vid, replica.type, size);
			map.put(entry.vid, entry);

			// add it to the Linked list, as the first unused entry
			if (unusedHeader == null) {
				unusedHeader = entry;
				entry.addAfter(header);
			} else {
				entry.addAfter(unusedHeader);
				unusedHeader = entry;
			}

			currCapacity += size;
		}
		// add replica to tuple space
		rout(replica);
		return true;
	}

	public synchronized DracoTuple rrdp(DracoTuple template)
	{
		if (template == null)
			throw new IllegalArgumentException();
		DracoTuple st = null;
		ITuple it = flatTuple(template);
		try {
			st = this.restoreDracoTuple(replicaTS.rdp(it));
		}
		catch (Exception e) {
			
			e.printStackTrace();
			st = null;
		}
		return st;
	}
	private DracoTuple rinp(DracoTuple template)
	{
		if (template == null)
			throw new IllegalArgumentException();
		DracoTuple st = null;
		ITuple it = flatTuple(template);
		try {
			st = this.restoreDracoTuple(replicaTS.inp(it));
		}
		catch (Exception e) {
			
			e.printStackTrace();
			st = null;
		}
		return st;
	}
	private DracoTuple[] ring(DracoTuple template)
	{
		if (template == null)
			throw new IllegalArgumentException();
		DracoTuple[] st = null;
		ITuple it = flatTuple(template);
		try {
			st = Util.convertITupleArrayToDracoTupleArray(replicaTS.ing(it));
		}
		catch (Exception e) {
			
			e.printStackTrace();
			st = null;
		}
		return st;
	}
	private DracoTuple rout(ShareTuple template)
	{
		if (template == null)
			throw new IllegalArgumentException();
		DracoTuple st = template;
		ITuple it = flatTuple(template);
		try {
			replicaTS.out(it);
		}
		catch (Exception e) {
			
			e.printStackTrace();
			st = null;
		}
		return st;
	}
	private DracoTuple[] routg(ShareTuple[] template)
	{
		if (template == null || template.length == 0)
			throw new IllegalArgumentException();
		DracoTuple[] st = template;
		ITuple[] it = Util.convertDracoTupleArrayToITupleArray(template);
		try {
			replicaTS.outg(it);
		}
		catch (Exception e) {
			
			e.printStackTrace();
			st = null;
		}
		return st;
	}
	
	

	private DracoTuple restoreDracoTuple(ITuple it) throws ClassNotFoundException  {
		DracoTuple st = null;

		if (it != null)

				st = DracoTuple.create(it);
	
		return st;
	}

	private ITuple flatTuple(DracoTuple st) {

		
		return st.toLimeTuple().limifyWrapper();
	}

	/**
	 * Add a replica array to local host replica the new replica to be added or
	 * not
	 * 
	 * @author Xing Jiankuan
	 */
	synchronized public boolean addReplicas(ShareTuple[] replicas) {

		HashMap<TupleID, ShareTuple> addedReplicas = new HashMap<TupleID, ShareTuple>();

		for (int i = 0; i < replicas.length; i++) {
			ShareTuple replica = replicas[i];
			if (replica == null)
				continue;

			assert (replica.isReplica);

			// check whether the repilca is already cached.
			TupleID vid = replica.vid;
			if (map.containsKey(vid))
				continue;

			if (!SystemTupleType.isSystemTupleType(replica.getType())) {

				// only replicas of non-system tuple type need replica
				// management
				int size = replica.getSize();

				if (size > maxCapacity) {
					System.err
							.println("Cannot create replica whose size is larger"
									+ " than maximal capacity");
					continue;
				}

				addedReplicas.put(replica.getVID(), replica);
				// remove replicas according to LRU
				while (currCapacity + size > maxCapacity) {
					Entry eldest = header.after;

					if (addedReplicas.containsKey(eldest.getVID())) {
						addedReplicas.remove(replica.getVID());
					}

					map.remove(eldest.vid);

					if (eldest == unusedHeader) {
						if (eldest.before == header) // the only one unused
							// entry is
							// removed
							unusedHeader = null;
						else
							unusedHeader = eldest.before;
					}
					eldest.remove();
					decreaseUsedCountByType(eldest);
					rinp(ShareTuple.createVIDTemplate(eldest.vid));
					currCapacity -= eldest.size;
				}

				// dts.getPolicyDecisionMaker().handleNewcoming(replica);

				// add it to hash map
				Entry entry = new Entry(replica.vid, replica.type, size);
				map.put(entry.vid, entry);

				// add it to the Linked list, as the first unused entry
				if (unusedHeader == null) {
					unusedHeader = entry;
					entry.addAfter(header);
				} else {
					entry.addAfter(unusedHeader);
					unusedHeader = entry;
				}

				currCapacity += size;
			} else {
				addedReplicas.put(replica.getVID(), replica);
			}

		}
		// add this whole replica array to tuple space
		ShareTuple[] st = new ShareTuple[addedReplicas.size()];
		addedReplicas.values().toArray(st);
		if (st.length != 0)
			routg(st);
		else
			System.out
					.println("The tuple array is null, so the operation is not executed!");
		return true;

	}

	

	/**
	 * Remove a replica
	 * 
	 * @param vid
	 *            the to be deleted replica's vid
	 * @return a copy of deleted replica
	 * @author Xing Jiankuan
	 */
	synchronized public ShareTuple removeReplica(TupleID vid) {
		DracoTuple dt = rinp(ShareTuple.createVIDTemplate(vid));
		if (dt == null)
			return null;

		ShareTuple st = null;
		if (!(dt instanceof ShareTuple)) {
			throw new IllegalArgumentException("The input vid does not"
					+ "represent a replica");
		}
		st = (ShareTuple) dt;
		if (!st.isReplica) {
			throw new IllegalArgumentException("The input vid does not"
					+ "represent a replica");
		}

		if (!SystemTupleType.isSystemTupleType(st.getType())) {
			// only non-system tuple type tuples need remove management
			// information
			Entry entry = get(vid);
			if (entry == null)
				return null;
			map.remove(vid);

			decreaseUsedCountByType(entry);

			entry.remove();
			resetUnusedHeader(entry);
			currCapacity -= entry.size;
		}

		return st;
	}

	synchronized public ShareTuple[] removeReplicas(TupleID[] vids) {
		LinkedList<ShareTuple> ret = new LinkedList<ShareTuple>();
		for (TupleID id : vids) {
			ShareTuple st = removeReplica(id);
			if (st != null)
				ret.add(st);
		}
		ShareTuple[] array = new ShareTuple[ret.size()];
		return ret.toArray(array);
	}

	synchronized public void clear() {

		// clear linked list
		header.before = header;
		header.after = header;
		unusedHeader = null;
		// clear hash map
		map.clear();

		// clear tuple space, that is, remove all the replicas
		dts.ing(ShareTuple.ALLREPLICA);
	}

	/**
	 * access a replica and put it to
	 * 
	 * @param vid
	 * @author Xing Jiankuan
	 */
	synchronized public void touch(ShareTuple ret) {
		TupleID vid = ret.getVID();
		Entry entry = (Entry) map.get(vid);
		if (entry != null) {

			if (!entry.isUsed())// increase rtu
			{
				dts.getPolicyDecisionMaker().handleTouch(ret);
			}
			increaseUsedCountByType(entry);

			resetUnusedHeader(entry); // reset unusedHeader

			entry.accessed(this); // put it to the head of list, and set used to
			// be true
			// System.out.println("*************");
			// System.out.println(entry.vid + "has been put to the head.");
			// System.out.println(this);
			// System.out.println("*************");
		}
	}

	/**
	 * access a replica and put it to
	 * 
	 * @param vid
	 * @author Xing Jiankuan
	 */
	synchronized public ShareTuple addReplicaAndTouch(ShareTuple replica) {

		assert (replica.isReplica);

		// check whether the repilca is already cached.
		TupleID vid = replica.vid;
		if (map.containsKey(vid))
			return null;

		if (!SystemTupleType.isSystemTupleType(replica.getType())) {
			// only replicas of non-system tuple type need replica management
			int size = replica.getSize();

			if (size > maxCapacity) {
				System.err.println("Cannot create replica whose size is larger"
						+ " than maximal capacity");
				return null;
			}

			// remove replicas according to LRU
			while (currCapacity + size > maxCapacity) {
				Entry eldest = header.after;
				map.remove(eldest.vid);
				if (eldest == unusedHeader) {
					if (eldest.before == header) // the only one unused entry is
						// removed
						unusedHeader = null;
					else
						unusedHeader = eldest.before;
				}
				eldest.remove();
				decreaseUsedCountByType(eldest);
				rinp(ShareTuple.createVIDTemplate(eldest.vid));
				currCapacity -= eldest.size;
			}

			// dts.getPolicyDecisionMaker().handleNewcoming(replica);

			// add it to hash map
			Entry entry = new Entry(replica.vid, replica.type, size);
			map.put(entry.vid, entry);

			// add it to the Linked list, as the first used entry
			if (entry != null) {

				if (!entry.isUsed())// increase rtu
				{
					dts.getPolicyDecisionMaker().handleTouch(replica);
				}
				increaseUsedCountByType(entry);

				// resetUnusedHeader(entry); //reset unusedHeader

				// entry.accessed(this); //put it to the head of list, and set
				// used to be true
				entry.addBefore(header);
				entry.used = true;
			}

			currCapacity += size;
		}
		// add replica to tuple space
		rout(replica);
		return replica;
	}

	/**
	 * access a replica array and put it to
	 * 
	 * @param vid
	 * @author Xing Jiankuan
	 */
	synchronized public boolean addReplicasAndTouch(ShareTuple[] replicas) {

		HashMap<TupleID, ShareTuple> addedReplicas = new HashMap<TupleID, ShareTuple>();

		for (int i = 0; i < replicas.length; i++) {
			ShareTuple replica = replicas[i];
			if (replica == null)
				continue;
			assert (replica.isReplica);

			// check whether the repilca is already cached.
			TupleID vid = replica.vid;
			if (map.containsKey(vid))
				continue;

			if (!SystemTupleType.isSystemTupleType(replica.getType())) {
				// only replicas of non-system tuple type need replica
				// management

				addedReplicas.put(replica.getVID(), replica);

				int size = replica.getSize();

				if (size > maxCapacity) {
					System.err
							.println("Cannot create replica whose size is larger"
									+ " than maximal capacity");
					return false;
				}

				// remove replicas according to LRU
				while (currCapacity + size > maxCapacity) {
					Entry eldest = header.after;
					if (addedReplicas.containsKey(eldest.getVID())) {
						addedReplicas.remove(replica.getVID());
					}
					map.remove(eldest.vid);
					if (eldest == unusedHeader) {
						if (eldest.before == header) // the only one unused
							// entry is removed
							unusedHeader = null;
						else
							unusedHeader = eldest.before;
					}
					eldest.remove();
					decreaseUsedCountByType(eldest);
					rinp(ShareTuple.createVIDTemplate(eldest.vid));
					currCapacity -= eldest.size;
				}

				// dts.getPolicyDecisionMaker().handleNewcoming(replica);

				// add it to hash map
				Entry entry = new Entry(replica.vid, replica.type, size);
				map.put(entry.vid, entry);

				// add it to the Linked list, as the first used entry
				if (entry != null) {

					if (!entry.isUsed())// increase rtu
					{
						dts.getPolicyDecisionMaker().handleTouch(replica);
					}
					increaseUsedCountByType(entry);
					// resetUnusedHeader(entry); //reset unusedHeader

					// entry.accessed(this); //put it to the head of list, and
					// set used to be true
					entry.addBefore(header);
					entry.used = true;
				}

				currCapacity += size;
			} else {
				addedReplicas.put(replicas[i].vid, replicas[i]);
			}

		}

		// add this whole replica array to tuple space
		ShareTuple[] st = new ShareTuple[addedReplicas.size()];
		addedReplicas.values().toArray(st);
		if (st.length != 0)
			routg(st);
		else
			System.out
					.println("The tuple array is null, so the operation is not executed!");
		return true;
	}

	

	private void resetUnusedHeader(Entry entry) {
		if (unusedHeader == entry) {
			if (unusedHeader.before == header)
				unusedHeader = null;
			else
				unusedHeader = unusedHeader.before;
		}
	}

	/**
	 * Find all the unused replica with the specified type
	 * 
	 * @param type
	 *            the specified type
	 * @return an array containing all vids with which tuples of the specified
	 *         type are unused. If nothing is found, return an empty arry
	 * @author Xing Jiankuan
	 */
	synchronized public TupleID[] getUnusedReplicaByType(TupleType type) {
		List unusedReplicas = null;

		if (unusedHeader == null)
			return new TupleID[0];
		// transverse all unused entries
		unusedReplicas = new ArrayList();
		for (Entry curr = unusedHeader; curr != header; curr = curr.before) {
			if (curr.type.equals(type))
				unusedReplicas.add(curr.vid);
		}

		TupleID[] ret = new TupleID[unusedReplicas.size()];
		for (int i = 0; i < ret.length; i++) {
			ret[i] = (TupleID) unusedReplicas.get(i);
		}
		return ret;
	}

	/*
	 * Increase the count of added tuple by 1
	 */
	private void increaseUsedCountByType(Entry entry) {
		if (!entry.used) {
			Long count;
			if (!typeAdded.containsKey(entry.getType()))
				count = new Long(0);
			else
				count = typeAdded.get(entry.getType());
			typeAdded.put(entry.getType(), ++count);
		}
	}

	/*
	 * Increase the count of deleted tuple by 1
	 */
	private void decreaseUsedCountByType(Entry entry) {
		if (entry.used) {
			Long count;
			if (!typeDeleted.containsKey(entry.getType()))
				count = new Long(0);
			else
				count = typeDeleted.get(entry.getType());
			typeDeleted.put(entry.getType(), ++count);

			// System.out.println("Now the deleted count is " + count);
		}
	}

	/*
	 * Return the count of added tuples for specific tuple type from the
	 * beginning of the time
	 * 
	 * @author Jinxue Zhang
	 */
	public long getAddedCountByType(TupleType type) {
		if (!typeAdded.containsKey(type)) {
			return 0;
		} else
			return typeAdded.get(type);
	}

	/*
	 * Return the count of deleted tuples for specific tuple type from the
	 * beginning of the time
	 * 
	 * @author Jinxue Zhang
	 */
	public long getDeletedCountByType(TupleType type) {
		if (!typeDeleted.containsKey(type)) {
			return 0;
		} else
			return typeDeleted.get(type);
	}

	/*
	 * Return the count of used tuples for specific tuple type from the
	 * beginning of the time
	 * 
	 * @author Jinxue Zhang
	 */
	public long getUsedCountByType(TupleType type) {
		if (!typeAdded.containsKey(type)) {
			return 0;
		} else
			return typeAdded.get(type) - typeDeleted.get(type);
	}

	public String toString() {
		StringBuffer sb = new StringBuffer();
		sb.append("Used:\n");
		Entry tail = (unusedHeader == null ? header : unusedHeader);
		for (Entry curr = header.before; curr != tail; curr = curr.before) {
			sb.append("\t" + curr.vid.toString() + "\n");
		}
		sb.append("Unused:\n");
		if (unusedHeader != null) {
			for (Entry curr = unusedHeader; curr != header; curr = curr.before) {
				sb.append("\t" + curr.vid.toString() + "\n");
			}
		}
		return sb.toString();
	}

	/**
	 * Entry for looped linked list
	 * 
	 * @author Xing Jiankuan
	 * 
	 */
	private static class Entry {
		Entry before, after;

		public Entry(TupleID vid, TupleType type, int size) {
			this.vid = vid;
			this.type = type;
			this.size = size;
			this.used = false;
		}

		private TupleID vid;
		private TupleType type;
		private boolean used;
		private int size;

		public TupleType getType() {
			return type;
		}

		public void setUsed() {
			used = true;
		}

		public boolean isUsed() {
			return used;
		}

		public int getSize() {
			return size;
		}

		public TupleID getVID() {
			return vid;
		}

		/**
		 * remove an entry from linked list
		 * 
		 * @author Xing Jiankuan
		 */
		public void remove() {
			before.after = this.after;
			after.before = this.before;

		}

		public void addBefore(Entry existingEntry) {
			this.before = existingEntry.before;
			this.after = existingEntry;
			before.after = this;
			after.before = this;
		}

		public void addAfter(Entry existingEntry) {
			this.after = existingEntry.after;
			this.before = existingEntry;
			before.after = this;
			after.before = this;
		}

		public void accessed(LocalReplicaRegistrar lrr) {
			remove();
			addBefore(lrr.header);
			used = true;
		}
	}

	public synchronized void rupdate(ShareTuple oldTuple, ShareTuple newTuple) {
		
		rinp(ShareTuple.createVIDTemplate(oldTuple.vid));
		rout(newTuple);
		
	}

	public synchronized DracoTuple[] rrdg(ShareTuple template) {
		if (template == null)
			throw new IllegalArgumentException();
		DracoTuple[] st = null;
		ITuple it = flatTuple(template);
		try {
			st = Util.convertITupleArrayToDracoTupleArray(replicaTS.rdg(it));
		}
		catch (Exception e) {
			
			e.printStackTrace();
			st = null;
		}
		return st;
	}

	
}